package com.webchat.connect.messagequeue.consumer.service;


import com.webchat.common.bean.APIResponseBean;
import com.webchat.common.bean.APIResponseBeanUtil;
import com.webchat.common.constants.ConnectConstants;
import com.webchat.common.enums.ChatMessageTypeEnum;
import com.webchat.common.enums.messagequeue.MessageQueueEnum;
import com.webchat.common.service.messagequeue.producer.MessageQueueProducer;
import com.webchat.common.util.JsonUtil;
import com.webchat.connect.service.AccountService;
import com.webchat.connect.websocket.handler.ChatWebSocketEndPointServletHandler;
import com.webchat.domain.dto.queue.ArticleDelayMessageDTO;
import com.webchat.domain.vo.request.mess.ChatMessageRequestVO;
import com.webchat.domain.vo.response.mess.ChatMessageResponseVO;
import com.webchat.domain.vo.response.mess.PublicAccountArticleMessageVO;
import com.webchat.domain.vo.response.publicaccount.ArticleBaseResponseVO;
import com.webchat.rmi.pgc.OfficialArticleClient;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;


/**
 * 公众号推文消费服务
 */
@Slf4j
@Service
public class ArticlePushConsumeService {

    @Autowired
    private AccountService accountService;

    @Autowired
    private OfficialArticleClient officialArticleClient;

    @Autowired
    private MessageQueueProducer<ChatMessageRequestVO, Long> messageQueueProducer;

    public void consume(String message) {

        /**
         * 发序列化解析消息对象
         */
        Set<String> messages = JsonUtil.fromJson(message, Set.class);
        Set<ArticleDelayMessageDTO> messageDtos =
                messages.stream().map(mess ->
                        JsonUtil.fromJson(mess, ArticleDelayMessageDTO.class))
                        .collect(Collectors.toSet());
        /**
         * 消息推送
         */
        for (ArticleDelayMessageDTO eventMessage : messageDtos) {
            this.doPush(eventMessage);
        }
    }

    private void doPush(ArticleDelayMessageDTO eventMessage) {

        String publicAccount = eventMessage.getPublicAccount();
        Long articleId = eventMessage.getArticleId();
        // 获取文章详情，走feign 远程获取
        PublicAccountArticleMessageVO articleVO = this.getArticleBaseFromRemote(articleId);
        if (articleVO == null) {
            return;
        }
        // 获取公众号所有订阅用户
        Set<String> userIds = accountService.getOfficialUserIds(publicAccount);
        if (CollectionUtils.isEmpty(userIds)) {
            return;
        }
        ChatMessageResponseVO chatMessageResponseVO = new ChatMessageResponseVO();
        chatMessageResponseVO.setPublicAccountArticle(articleVO);
        chatMessageResponseVO.setSenderId(publicAccount);
        chatMessageResponseVO.setType(ChatMessageTypeEnum.PUBLIC_ACCOUNT_ARTICLE.getType());

        Set<String> bizCodes = ConnectConstants.ConnectBiz.getBizCode(ConnectConstants.BizEnum.CHAT);
        for (String bizCode : bizCodes) {
            Map<String, WebSocketSession> userWsMap = ChatWebSocketEndPointServletHandler.getSessions(bizCode, userIds);
            for (String userId : userIds) {
                WebSocketSession wsSession = userWsMap.get(userId);
                if (wsSession == null || !wsSession.isOpen()) {
                    continue;
                }
                chatMessageResponseVO.setReceiverId(userId);
                try {
                    /**
                     * 公众号文章推送
                     */
                    wsSession.sendMessage(new TextMessage(JsonUtil.toJsonString(chatMessageResponseVO)));
                    /**
                     *  推送数据持久化
                     */
                    this.persistentArticlePushMessage(publicAccount, userId, articleId);
                } catch (IOException e) {
                    // 不能抛异常出去，不能阻塞其他用户的推送
                    log.error("公众号文章推送异常 =====> publicAccount:{}, userId:{}, articleId:{}",
                            publicAccount, userId, articleId, e);
                }

            }
        }
    }


    /**
     * 公众号推送消息持久化
     *
     */
    private void persistentArticlePushMessage(String officialAccount, String userId, Long articleId) {

        ChatMessageRequestVO persistentMessage = new ChatMessageRequestVO();
        persistentMessage.setType(ChatMessageTypeEnum.PUBLIC_ACCOUNT_ARTICLE.getType());
        persistentMessage.setSenderId(officialAccount);
        persistentMessage.setReceiverId(userId);
        persistentMessage.setMessage(String.valueOf(articleId));
        /**
         * 走统一的消息持久化处理服务（服用了对话场景持久化服务）
         *
         */
        messageQueueProducer.send(MessageQueueEnum.QUEUE_PERSISTENT_MESSAGE, persistentMessage);
    }

    /**
     * 获取文章详情，这里不需要返回正文（降低相应网络传输包大小，节约流量成本，提高接口性能）
     *
     * @param id
     * @return
     */
    private PublicAccountArticleMessageVO getArticleBaseFromRemote(Long id) {
        APIResponseBean<ArticleBaseResponseVO> responseBean = officialArticleClient.detail(id, false);
        if (APIResponseBeanUtil.isOk(responseBean)) {
            ArticleBaseResponseVO baseResponseVO = responseBean.getData();
            PublicAccountArticleMessageVO publicAccountArticleMessageVO = new PublicAccountArticleMessageVO();
            publicAccountArticleMessageVO.setArticleId(baseResponseVO.getId());
            publicAccountArticleMessageVO.setTime(baseResponseVO.getPlanPushTime());
            publicAccountArticleMessageVO.setTitle(baseResponseVO.getTitle());
            publicAccountArticleMessageVO.setCover(baseResponseVO.getCover());
            publicAccountArticleMessageVO.setDescription(baseResponseVO.getDescription());
            publicAccountArticleMessageVO.setRedirectUrl(baseResponseVO.getRedirectUrl());
            return publicAccountArticleMessageVO;
        }
        log.error("远程获取公众号文章详情失败 =====> id:{}, responseBean:{}", id, JsonUtil.toJsonString(responseBean));
        return null;
    }

}
